package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo4RePartition {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    //读取数据
    val linesDS: RDD[String] = sc.textFile("data/students.txt")

    println(s"linesDS分区数：${linesDS.getNumPartitions}")

    /**
     * repartition: 重分区，会产生shuffle
     */
    val rdd2: RDD[String] = linesDS.repartition(10)

    println(s"rdd2分区数：${rdd2.getNumPartitions}")

    /**
     * coalesce: 重分区，可以设置是否产生shuffle
     * 有shuffle:可以用于增加分区，也可以用于减少分区
     * 没有shuffle: 只能用于减少分区不能用于增加分区
     */
    val rdd3: RDD[String] = rdd2.coalesce(5, false)
    println(s"rdd3分区数：${rdd3.getNumPartitions}")

    rdd3
      .repartition(1000)
      //.saveAsTextFile("data/repartition")

    //读取小文件
    val rdd4: RDD[String] = sc.textFile("data/repartition")
    println(s"rdd4分区数：${rdd4.getNumPartitions}")

    //当每一个分区中数据很少的时候可以使用coalesce合并分区，减少小文件
    // 使用coalesce合并小文件, 不能产生shuffle
    val rrd5: RDD[String] = rdd4.coalesce(2)
    println(s"rrd5分区数：${rrd5.getNumPartitions}")
    rrd5.saveAsTextFile("data/coalesce")
  }

}
